package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TradeProvinceOrderWindow;
import com.atguigu.func.DimAsyncJoinFunction;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MyClickHouseUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
//数据流：web/app -> mysql -> maxwell -> kafka(ods) -> flinkApp(订单明细关联后的事实表) -> kafka(dwd) ->flinkApp -> Clickhouse(DWS)
//程 序：Mock      ->mysql -> maxwell -> kafka(zk) -> DwdTradeOrderDetail ->kafka(zk) ->Dws10TradeProvinceOrderWindow(redis,hdfs,zk,hbase,phoenix)->clickhouse(zk)
/*
10.10 交易域省份粒度下单各窗口汇总表
（按照事件时间开窗，一个订单会有多个订单明细，同一个订单的订单明细的事件时间肯定会一样，会进到同一个窗口，所以我们只需要对同一个窗口对order_id进行去重即可）
10.10.1 主要任务
从 Kafka 读取业务数据，筛选订单表数据，统计各省份各窗口订单数和订单金额，将数据写入 ClickHouse 交易域省份粒度下单各窗口汇总表。

 */
//todo 1.获取执行环境
//todo 2.读取dwd层订单明细主题创建流（dwd层订单明细时通过left join产生的，会产生左 null，null，左右）
//todo 3.过滤null值并转化为json对象
//todo 4.按照订单明细id分组，去重数据
//todo 5.将数据转换为JavaBean对象
//todo 6.提取时间戳生成watermark
//todo 7.按照省份id分组进行开窗聚合
//todo 8.关联省份表补充维度信息(聚合后再关联，数据量就会变少很多)
//todo 9.将数据写到clickhouse
//todo 10.启动任务
public class Dws10TradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 生产环境一定要写，测试注释掉，否则每次测试都得开hdfs
//        需要从checkpoint或savepoint启动程序
//        //2.1 开启checkpoint，每隔5s钟做一次ck，并指定ck的一致性语义
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);//exactly once：默认barrier对齐
//        //2.2 设置超时时间为1min
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);//设置超时时间设置checkpoint的超时时间为1min，是指做一次checkpoint的时间；如果超时则认为本次checkpoint失败，这个checkpoint就丢了，继续一下一次checkpoint即可
//        //2.3设置两次重启的最小时间间隔为3s
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        //2.4设置任务关闭的时候保留最后一次ck数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        //2.5 指定从ck自动重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1L),Time.minutes(1L)
//        ));
//        //2.6 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());//本地状态位置
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/flinkCDC/220828"
//        );//checkpoint状态位置
//        //2.7 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME","atguigu");

        //todo 2.读取dwd层订单明细主题创建流（dwd层订单明细时通过left join产生的，会产生左 null，null，左右）
        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", "province_order1_220828").setStartFromEarliest());

        //todo 3.过滤null值并转化为json对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                }
            }
        });
//        jsonObjDS.print("jsonObjDS>>>>>>");

        //todo 4.按照订单明细id分组，去重数据(主键本来唯一，但是由于left join产生多个)
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.keyBy(json -> json.getString("id"))
                .filter(new RichFilterFunction<JSONObject>() {
                    //todo 过滤掉重复的order_detail_id，定义状态，如果为null，就保留，不为null，就不要，并更新状态为随便的一个值
                    private ValueState<String> valueState;

                    //open方法里对状态初始化
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<String>("order-detail-state", String.class);
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(10))
                                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                                .build();
                        stateDescriptor.enableTimeToLive(ttlConfig);
                        valueState = getRuntimeContext().getState(stateDescriptor);

                    }

                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        //获取状态
                        String stateValue = valueState.value();

                        //过滤数据
                        if (stateValue == null) {
                            valueState.update("1");
                            return true;//将同一个order_detail_id来的第一条数据返回，其他的重复的过滤
                        } else {
                            return false;
                        }
                    }
                });

//        filterDS.print("filterDS>>>>");

        //todo 5.将数据转换为JavaBean对象

        SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceOrderDS = filterDS.map(new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
            @Override
            public TradeProvinceOrderWindow map(JSONObject value) throws Exception {

                HashSet<String> orderIds = new HashSet<>();
                orderIds.add(value.getString("order_id"));

                return TradeProvinceOrderWindow.builder()
                        .provinceId(value.getString("province_id"))
                        .orderIdSet(orderIds)//多个订单明细id（id不相同）的order_id相同，order_id进到Set集合里会自动去重，对应的事件时间是一样的，会进到一个窗口
                        .orderAmount(value.getBigDecimal("split_total_amount"))//原始金额，一定是有值的，不用判断是否为null
//                        .ts(value.getLong("create_time"))
                        .ts(DateFormatUtil.toTs(value.getString("create_time"),true))
                        .build();
            }
        });

//        provinceOrderDS.print("provinceOrderDS>>>>");
        //todo 6.提取时间戳生成watermark
        SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceOrderWithWMDS = provinceOrderDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindow>() {
                            @Override
                            public long extractTimestamp(TradeProvinceOrderWindow element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        //todo 7.按照省份id分组进行开窗聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> reducedDS = provinceOrderWithWMDS.keyBy(s -> s.getProvinceId())
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {

                        //将窗口里来的一条条javabean里的oderSet里的数据聚合成一个set
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());//将v2也就是后面数据的orderSet里的订单id添加进第一条数据的订单id里，并去重了，去重后的订单id都不相同
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;

                    }
                }, new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
                        //获取数据
                        TradeProvinceOrderWindow next = input.iterator().next();

                        //todo 补充去重后的订单数(也就是开窗聚合后的唯一的数据的orderSet的size大小)
                        next.setOrderCount((long) next.getOrderIdSet().size());
                        //补充信息
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setTs(System.currentTimeMillis());

                        //一定不能忘了要把数据收集发送给下游
                        out.collect(next);


                    }
                });

//        provinceOrderWithWMDS.print("provinceOrderWithWMDS>>>>>>");//有数据
        reducedDS.print("reducedDS>>>>>>");//没有数据

        //todo 8.关联省份表补充维度信息(聚合后再关联，数据量就会变少很多)
        SingleOutputStreamOperator<TradeProvinceOrderWindow> resultDS = AsyncDataStream.unorderedWait(reducedDS, new DimAsyncJoinFunction<TradeProvinceOrderWindow>("DIM_BASE_PROVINCE") {
            @Override
            public String getKey(TradeProvinceOrderWindow input) {
                return input.getProvinceId();//获取上游设置进去的省份id用来关联省份维表
            }

            @Override
            public void join(TradeProvinceOrderWindow input, JSONObject dimInfo) {
                input.setProvinceName(dimInfo.getString("NAME"));//获取phoenix查询出来的省份名，要看sink_columns的字段

            }
        }, 60, TimeUnit.SECONDS);

        resultDS.print("即将写入clickhouse的数据,跳过oderIdSet");

        //todo 9.将数据写到clickhouse
        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)"));

        //todo 10.启动任务
        env.execute("Dws10TradeProvinceOrderWindow");

    }
}
